package com.clw.utils;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.phoenix.jdbc.PhoenixDriver;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Set;

public class DimSinkHbase extends RichSinkFunction<JSONObject> {

    Connection connection = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        PhoenixDriver t;
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        connection = DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181");
    }




    /**
     * 生成语句提交hbase
     * @param jsonObject
     * @param context
     * @throws Exception
     */
    @Override
    public void invoke(JSONObject jsonObject, Context context) throws Exception {

        String tableName = jsonObject.getString("sink_table");
        String type = jsonObject.getString("type");
        JSONObject dataJsonObj = jsonObject.getJSONObject("data");

        //this.checkTable(tableName,jsonObject.getString("sink_fields"),"id","ext");
        if (dataJsonObj != null && dataJsonObj.size() > 0) {
            String strSql="";
            if("insert".equals(type) || "update".equals(type)) {
                // 生成添加或者修改的的sql 语句
                strSql = genUpsertSql(tableName.toUpperCase(), jsonObject.getJSONObject("data"));
            }else if("delete".equals(type)){
                strSql = genDelSql(tableName.toUpperCase(), jsonObject.getJSONObject("data"),jsonObject.getString("pk"));
            }


            try {
                System.out.println(strSql);
                PreparedStatement ps = connection.prepareStatement(strSql);
                ps.executeUpdate();
                connection.commit();
                ps.close();
            } catch (Exception e) {
                e.printStackTrace();
                throw new RuntimeException("执行sql失败！");
            }
        }
    }

    public String genUpsertSql(String tableName, JSONObject jsonObject) {
        Set<String> fields = jsonObject.keySet();
        String upsertSql = "upsert into " + "MYSHEMA1" + "." + tableName + "(" + StringUtils.join(fields, ",") + ")";
        String valuesSql = " values ('" + StringUtils.join(jsonObject.values(), "','") + "')";
        return upsertSql + valuesSql;
    }
    public String genDelSql(String tableName, JSONObject jsonObject,String pk) {
        Set<String> fields = jsonObject.keySet();
        // 获取主键的值
        String keyVal = jsonObject.getString(pk);
        String strSql = "delete from  " + "MYSHEMA1" + "." + tableName + "  where " + pk + "'" + keyVal + "'";
        return strSql;
    }

}
